#ifndef _M_MSG_H_
#define _M_MSG_H_

#include"../mqcommon/mq_helper.hpp"
#include"../mqcommon/mq_msg.pb.h"
#include<iostream>
#include<list>
#include<unordered_map>
#include<mutex>
#include<memory>

namespace xiang11mq
{
    #define DATAFILE_SUBFIX ".mqd"      //后缀
    #define TMPFILE_SUBFIX ".mqd.tmp"   //临时文件后缀
    //proto 头文件的智能指针自己创建
    using MessagePtr = std::shared_ptr<xiang11mq::Message>;

//消息的持久化不在数据库 在文件上 因为消息的长度可能非常长 而且不提供查询消息的功能
//消息的持久化管理
class MessageMapper
{
public:
    MessageMapper(std::string &basedir, const std::string &qname)
        :_qname(qname)
    {
        if(basedir.back() != '/')
            basedir.push_back('/');
        _datafile = basedir + qname + DATAFILE_SUBFIX;
        _tmpfile = basedir + qname + TMPFILE_SUBFIX;
        if(FileHelper(basedir).exists() == false)
        {
            assert(FileHelper::createDirectory(basedir));
        }
        createMsgFile();
    }
    //创建消息文件
    bool createMsgFile()
    {
        bool ret = FileHelper::createFile(_tmpfile);
        if(ret == false)
        {
            DLOG("创建队列临时文件 %s 失败!", _tmpfile.c_str());
            return false;
        }
        if(FileHelper(_datafile).exists() == true)
        {
            return true;
        }
        ret = FileHelper::createFile(_datafile);
        if(ret == false)
        {
            DLOG("创建队列文件 %s 失败!", _datafile.c_str());
            return false;
        }

        return true;
    }
    //删除消息文件
    void removeMsgFile()
    {
        //不要删除目录 可能还有其他消息文件
        FileHelper::removeFile(_datafile);
        FileHelper::removeFile(_tmpfile);
    }
    //插入消息  必须传引用
    bool insert(MessagePtr &msg)
    {
        return insert(_datafile, msg);
    }
    //删除消息
    bool remove(MessagePtr &msg)
    {
        //1.将msg中的有效标志位 存储为 '0'
        msg->mutable_payload()->set_valid("0");
        //2.对msg进行序列化
        std::string body = msg->payload().SerializeAsString();
        if(body.size() != msg->length())
        {
            DLOG("不能修改文件中的信息 新生成的数据与原数据长度不一致");
            return false;
        }
        //3.覆盖原有数据
        FileHelper helper(_datafile);
        bool ret = helper.write(body.c_str(), msg->offset(), body.size());
        if(ret == false)
        {
            DLOG("向队列数据文件写入失败！");
            return false;
        }
        //DLOG("确认消息后, 删除持久化消息成功 %s", msg->payload().body().c_str());
        return true;
    }
    //历史数据的恢复和垃圾回收 //Garbage Collection
    std::list<MessagePtr> gc()
    {
        std::list<MessagePtr> result;  
        //1.加载出文件中所有的有效数据  存储格式 4字节长度|数据|4字节长度|数据......
        bool ret = load(result);
        if(ret == false)
        {
            DLOG("加载有效消息失败");
            return result;
        }
        //DLOG("垃圾回收或数据恢复 得到有效消息数量 %lu", result.size());
        //2.将有效数据 进行序列化存储到临时文件中
        for(auto &msg : result)
        {
            DLOG("向临时文件写入数据: %s", msg->payload().body().c_str());
            ret = insert(_tmpfile, msg);
            if(ret == false)
            {
                DLOG("向临时文件写入消息失败！");
                return result;
            }
        }
        //DLOG("垃圾回收或数据恢复 数据写入临时文件后 临时文件大小 %lu", FileHelper(_tmpfile).size());
        //3.删除源文件
        ret = FileHelper::removeFile(_datafile);
        if (ret == false)
        {
            DLOG("删除源文件失败！");
            return result;
        }
        // 4.修改临时文件的文件名 为源文件名称
        ret = FileHelper(_tmpfile).rename(_datafile);   //写入文件会自动创建tmpfile 用就有
        if(ret == false)
        {
            DLOG("修改临时文件名称失败！");
            return result;
        }
        //5.返回新的有效数据
        return result;
    }

private:
    //加载历史消息
    bool load(std::list<MessagePtr> &result)
    {
        FileHelper helper(_datafile);
        size_t offset = 0;
        size_t msg_size = 0;
        size_t fsize = helper.size();
        bool ret;
        //DLOG("准备加载历史数据时文件大小 %lu", helper.size());
        while(offset < fsize)
        {
            ret = helper.read((char *)&msg_size, offset, sizeof(size_t));
            offset += sizeof(size_t);
            if(!ret)
            {
                DLOG("读取消息长度失败");
                return false;
            }
            std::string msg_body(msg_size, '\0');
            ret = helper.read(&msg_body[0], offset, msg_size);
            if(!ret)
            {
                DLOG("读取消息数据失败");
                return false;
            }
            offset += msg_size;

            MessagePtr msgp = std::make_shared<Message>();
            msgp->mutable_payload()->ParseFromString(msg_body);
            //如果是无效消息 直接处理下一个 不然保存起来
            if(msgp->payload().valid() == "0") 
            {
                DLOG("加载到无效数据 %s", msgp->payload().body().c_str());
                continue;
            }
            else if(msgp->payload().valid() == "1") 
            {
                DLOG("加载到有效数据 %s", msgp->payload().body().c_str());
            }
            result.push_back(msgp);
            
        }     
        return true;
    }
    //向文件写入msg
    bool insert(const std::string &filename, MessagePtr &msg)
    {
        //新增数据是添加在文件末尾的
        //1.进行消息的序列化， 获取格式化后的信息
        std::string body = msg->payload().SerializeAsString();
        //2.获取文件长度
        FileHelper helper(filename);
        size_t fsize = helper.size();
        size_t msg_size = body.size();
        bool ret;
        //3.将数据写入文件指定位置  写入逻辑: 先写入长度 再写入数据
        ret = helper.write((char*)&msg_size, fsize, sizeof(size_t));
        if(ret == false)
        {
            DLOG("向队列数据文件写入数据长度失败");
            return false;
        }
        ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
        if(ret == false)
        {
            DLOG("向队列数据文件写入数据失败");
            return false;
        }
        //4.更新msg中的实际存储信息
        msg->set_offset(fsize + sizeof(size_t));
        msg->set_length(body.size());
        return true;
    }
private:
    std::string _qname;
    std::string _datafile;      //磁盘上的文件
    std::string _tmpfile;       //临时文件
}; 
//队列 单元消息类
class QueueMessage
{
public:
    using ptr = std::shared_ptr<QueueMessage>;
    QueueMessage(std::string &basedir, const std::string &qname)
        :_qname(qname)
        ,_vaild_count(0)
        ,_total_count(0)
        ,_mapper(basedir, qname)
    {   

    }
    //恢复历史消息
    void recovery()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        // 回复历史消息
        _msgs = _mapper.gc();
        for (auto &msg : _msgs)
        {
            _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
        }
        _vaild_count = _total_count = _msgs.size();
    }
    //插入消息 消息的基础属性 + 消息的内容 + (是否有效) 构建消息对象
    bool insert(const BasicProperties* bp, const std::string &body, bool queue_is_durable)
    {
        //1.构造消息对象
        MessagePtr msg = std::make_shared<Message>();
        msg->mutable_payload()->set_body(body);
        
        if(bp != nullptr)   //客户端也设置了属性
        {
            DeliverMode mode = queue_is_durable ? bp->deliver_mode() : DeliverMode::UNDURABLE;
            msg->mutable_payload()->mutable_properties()->set_id(bp->id());
            msg->mutable_payload()->mutable_properties()->set_deliver_mode(mode);
            msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
        }
        else    //客户端只是发了条消息 没有设置属性
        {
            DeliverMode mode = queue_is_durable ? DeliverMode::DURABLE : DeliverMode::UNDURABLE;
            msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
            msg->mutable_payload()->mutable_properties()->set_deliver_mode(mode);
            msg->mutable_payload()->mutable_properties()->set_routing_key("");
        }
        std::unique_lock<std::mutex> lock(_mutex);
        //2.判断消息是否要持久化
        if(msg->payload().properties().deliver_mode() == DeliverMode::DURABLE)
        {
            msg->mutable_payload()->set_valid("1");     //设置持久化
            //3.持久化存储
            bool ret = _mapper.insert(msg);
            if(ret == false)
            {
                DLOG("持久化存储消息 %s 失败了", body.c_str());
                return false;
            }
            _vaild_count += 1; //持久化信息数量 + 1；
            _total_count += 1;
            _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
        }
        //4.内存的管理
        _msgs.push_back(msg);
        return true;
    }
    //获取队首消息-->推送消息 放入待确认hash里
    MessagePtr front()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        if (_msgs.size() == 0)
        {
            return MessagePtr();
        }
        //获取一条队首消息 从_msgs中取出
        MessagePtr msg = _msgs.front();
        _msgs.pop_front();
        //向待确认hash中添加一份 成功确认了再删除 防止消息处理失败
        _waitack_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
        return msg;
    }
    //删除/确认消息  每次确认消息后 判断是否要垃圾回收
    bool remove(const std::string &msg_id)
    {
        std::unique_lock<std::mutex> lock(_mutex);
        //1.从待确认队列中寻找消息
        auto it = _waitack_msgs.find(msg_id);
        if(it == _waitack_msgs.end())
        {
            DLOG("没有找到要删除的消息 %s", msg_id.c_str());
            return true;
        }
        //2.根据消息的持久化模式 决定是否删除持久化信息
        if(it->second->payload().properties().deliver_mode() == DeliverMode::DURABLE)
        {
            //3.删除持久化消息   -->从文件中拿到消息 valid改为0再写回去 gc的时候才会扔掉
            _mapper.remove(it->second);
            _durable_msgs.erase(msg_id);
            _vaild_count -= 1;  //吃持久化消息数量减一
            //_total_count -= 1;    total不减 计算消息有效比例
            gc();               //内部判断消息回收
        }
        //4.删除内存消息
        _waitack_msgs.erase(msg_id);
        //DLOG("确认消息后，删除消息的管理成功: %s", it->second->payload().body().c_str());
        return true;
    }
    //可获取消息数量
    size_t getable_count()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        return _msgs.size();
    }
    //总消息数量
    size_t total_count()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        return _total_count;
    }
    //持久化消息数量
    size_t durable_count()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        return _durable_msgs.size();
    }
    //待确认消息数量
    size_t waitack_count()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        return _waitack_msgs.size();
    }
    //清理
    void clear()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        _mapper.removeMsgFile();
        _msgs.clear();
        _durable_msgs.clear();
        _waitack_msgs.clear();
        _vaild_count = _total_count = 0;
    }
private:
    bool GCcheck()
    {
        //持久化消息总量大于2000 且有效比例低于50% 则需要清理
        if(_total_count > 2000 && _vaild_count * 10 / _total_count < 5)
        {
            return true;
        }
        return false;
    }
    void gc()
    {
        //1.进行垃圾回收 获取有效的消息链表
        if(GCcheck() == false) return;
        std::list<MessagePtr> msgs = _mapper.gc();
        for(auto &msg : msgs)
        {
            auto it = _durable_msgs.find(msg->payload().properties().id());
            if(it == _durable_msgs.end())
            {
                DLOG("垃圾回收中 有持久化消息丢失");
                _msgs.push_back(msg);           //重新推送到链表末尾
                //重新插入到durable_msgs    
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                continue;
            }
            //2.更新每一条消息的实际存储位置
            it->second->set_offset(msg->offset());
            it->second->set_length(msg->length());
        }
        //3.更新当前有效消息数量 & 总持久化消息数量
        _vaild_count = _total_count = msgs.size();
    }
private:
    std::mutex _mutex;
    std::string _qname;                                         //队列名
    size_t _vaild_count;                                        // 持久化/有效消息数量 = _durable_msgs.size()
    size_t _total_count;                                        //总持久化消息数量
    MessageMapper _mapper;                                      //消息持久化管理类

    std::list<MessagePtr> _msgs;                                //待推送消息    内存中所有不持久化和持久化的消息
    std::unordered_map<std::string, MessagePtr> _durable_msgs;  //内存中持久化消息hash  内存中所有持久化的消息
    std::unordered_map<std::string, MessagePtr> _waitack_msgs;  //待确认消息hash
};

//队列 单元消息管理类
class MessageManager
{
public:
    using ptr = std::shared_ptr<MessageManager>;
    MessageManager(const std::string &basedir)
        :_basedir(basedir)
    {}
    //初始化队列消息
    void initQueueMessage(const std::string &qname)
    {
        // manager 和 mapper 中都有锁 为了避免锁冲突 浪费资源 分开作用域
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it != _queue_msgs.end())
            {
                DLOG("初始化队列消息时 队列%s已经存在", qname.c_str());
                return;
            }
            qmp = std::make_shared<QueueMessage>(_basedir, qname);
            _queue_msgs.insert(std::make_pair(qname, qmp));
        }
        qmp->recovery();
    }
    //销毁队列消息
    void destroyQueueMessage(const std::string &qname)
    {   
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("要销毁的队列消息不存在!");
                return;
            }
            qmp = it->second;
            _queue_msgs.erase(qname);
        }
        qmp->clear();
    }
    //向某个队列新增消息     队列名 消息基本属性 消息主体 是否需要持久化
    bool insert(const std::string &qname, BasicProperties *bp, const std::string &body, bool queue_is_durable)
    {
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("向队列%s新增消息失败: 没有找到消息管理句柄", qname.c_str());
                return false;
            }
            qmp = it->second;
       }
       return qmp->insert(bp, body, queue_is_durable);
    }
    //获取指定队列的队首消息
    MessagePtr front(const std::string &qname)
    {
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("获取队列%s队首消息失败: 没有找到消息管理句柄", qname.c_str());
                return MessagePtr();
            }
            //DLOG("成功找到了队列%s消息管理句柄", qname.c_str());
            qmp = it->second;
        }
        return qmp->front();
    }
    //确认/删除某个队列某个id的消息
    bool ack(const std::string &qname, const std::string &msg_id)
    {
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("确认队列%s 消息%s 失败：没有找到消息管理句柄", qname.c_str(), msg_id.c_str());
                return false;
            }
            qmp = it->second;
        }
        return qmp->remove(msg_id); //确认就是删除
    }
    //某个队列可获取消息数量
    size_t getable_count(const std::string &qname)
    {
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("获取队列%s 待推送消息数量失败：没有找到消息管理句柄", qname.c_str());
                return 0;
            }
            qmp = it->second;
        }
        return qmp->getable_count(); 
    }
    //某个队列总消息数量
    size_t total_count(const std::string &qname)
    {
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("获取队列%s 总消息数量失败：没有找到消息管理句柄", qname.c_str());
                return 0;
            }
            qmp = it->second;
        }
        return qmp->total_count(); 
    }
    //某个队列持久化消息数量
    size_t durable_count(const std::string &qname)
    {
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("获取队列%s 持久化消息数量失败：没有找到消息管理句柄", qname.c_str());
                return 0;
            }
            qmp = it->second;
        }
        return qmp->durable_count(); 
    }
    //某个队列待确认消息数量
    size_t waitack_count(const std::string &qname)
    {
        QueueMessage::ptr qmp;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msgs.find(qname);
            if (it == _queue_msgs.end())
            {
                DLOG("获取队列%s 待确认消息数量失败：没有找到消息管理句柄", qname.c_str());
                return 0;
            }
            qmp = it->second;
        }
        return qmp->waitack_count(); 
    }
    //清理
    void clear()
    {
        std::unique_lock<std::mutex> lock(_mutex);
        for(auto &qmsg : _queue_msgs)
            qmsg.second->clear();
    }

private:
    std::mutex _mutex;
    std::string _basedir;                                               //所有消息文件存放的目录
    std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;     //所有 队列单元消息
};


}


#endif